package io2.nio.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class EventLoopDemo {
    // 监听 Accept 线程数
    static final int MAIN_THREAD_NUM = 1;
    // I/O 线程数
    static final int IO_THREAD_NUM = 4;
    // accept 线程组
    static final EventLoop[] mainGroup = new EventLoop[MAIN_THREAD_NUM];
    // io 线程组
    static final EventLoop[] ioGroup = new EventLoop[IO_THREAD_NUM];
    // 处理业务的线程池
    static final ExecutorService executorService = Executors.newCachedThreadPool();
    // http响应数据
    static final byte[] RS = "HTTP/1.1 200 OK\r\nContent-length:13\r\n\r\nHello, World!\r\n".getBytes();

    /**
     * 初始化线程组
     */
    static void initGroup() {
        // 初始化accept线程
        for (int i = 0; i < MAIN_THREAD_NUM; i++) {
            mainGroup[i] = new EventLoop("main-group-" + i) {
                private AtomicInteger count = new AtomicInteger();
                /**
                 * 处理accept事件
                 * @param channel
                 * @param key
                 */
                @Override
                public void handle(Object channel, SelectionKey key) {
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) channel;
                    try {
                        SocketChannel accept = serverSocketChannel.accept();
                        if(accept != null) {
                            accept.configureBlocking(false);
                            log("新建连接：" + accept.getRemoteAddress());
                            // 选择一个io线程，这里使用轮训
                            EventLoop eventLoop = ioGroup[count.incrementAndGet() % IO_THREAD_NUM];
                            // 启动io线程
                            eventLoop.doStart();
                            // 注册到io线程上
                            SelectionKey selectionKey = eventLoop.register(accept);
                            // 订阅读取事件
                            selectionKey.interestOps(SelectionKey.OP_READ);
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            };
        }
        for (int i = 0; i < IO_THREAD_NUM; i++) {
            ioGroup[i] = new EventLoop("io-group-" + i) {
                /**
                 * 处理io事件
                 * @param channel
                 * @param key
                 */
                @Override
                public void handle(Object channel, SelectionKey key) {
                    SocketChannel socketChannel = (SocketChannel) channel;
                    log("有数据可以读取了");
                    final ByteBuffer buffer = ByteBuffer.allocate(1024);
                    try {
                        if(socketChannel.read(buffer) <= 0) {
                            log("没有数据可以读取");
                            return;
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                        log("读取数据出错");
                        return;
                    }
                    executorService.execute(() -> {
                        buffer.flip();
                        byte[] bytes = new byte[buffer.remaining()];
                        buffer.get(bytes);
                        log("收到数据：" + new String(bytes));
                        try {
                            socketChannel.write(ByteBuffer.wrap(RS));
//                            key.cancel();
                            socketChannel.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        System.out.println("响应数据发送完毕");
                    });
                }
            };
        }
    }

    /**
     * 日志输出
     * @param str
     */
    static void log(String str) {
        System.out.println("[" + Thread.currentThread().getName() + "] - " + str);
    }

    /**
     * 初始化服务器端通道
     */
    static ServerSocketChannel initServerChannel() {
        ServerSocketChannel serverSocketChannel = null;
        try {
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            // 启动accept线程
            mainGroup[0].doStart();
            // 把serverSocketChannel注册到accept线程上
            SelectionKey selectionKey = mainGroup[0].register(serverSocketChannel);
            // 订阅 accept 事件
            selectionKey.interestOps(SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return serverSocketChannel;
    }

    /**
     * 绑定端口
     */
    static void doBind(ServerSocketChannel serverSocketChannel) {
        try {
            serverSocketChannel.bind(new InetSocketAddress(8080));
        } catch (IOException e) {
            e.printStackTrace();
        }
        log("服务启动成功， 监听8080端口");
    }

    public static void main(String[] args) {
        initGroup();
        doBind(initServerChannel());
    }
}
